package com.lzy.gmall.realtime.app.dws;

import com.lzy.gmall.realtime.bean.T1;
import com.lzy.gmall.realtime.utils.MyClickHouseUtil;
import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/**
*
*
*                       指标一 加购人数
*
* */
public class Dws_Number_of_additional_purchases {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String topic = "dwd_order_pre";
        String groupId = "dws_add_to_cart_user_count";

        // 定义表结构
        tableEnv.executeSql("CREATE TABLE DwsShopSource (" +
                "   user_id STRING," +
                "   sku_id bigint," +
                "   sku_name STRING," +
                "   create_time STRING," +
//                将 row_op_ts通过REPLACE转换为 Flink适用戳类型。
                "   row_op_ts STRING," +
                "  `time_ltz` AS TO_TIMESTAMP_LTZ(" +
                "    UNIX_TIMESTAMP(REPLACE(row_op_ts, ' ', 'T'), 'yyyy-MM-dd''T''HH:mm:ss') * 1000," +
                "    3" +
                "  )," +
                "  WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND" +//设置水位线5秒延迟数据
                ")" + MyKafkaUtil.getKafkaDDL(topic, groupId));


        Table t0 = tableEnv.sqlQuery("SELECT " +
                "  DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS stt," +
                "  DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS edt, " +
                "  sku_id, " +
                "  sku_name, " +  // 添加商品名称
                "  count(DISTINCT user_id) AS add_to_cart_users " +
                "FROM TABLE(" +
//                      滚动窗口每 10 秒生成一个不重叠的窗口统计窗口内的数据
                "  TUMBLE(TABLE DwsShopSource, DESCRIPTOR(time_ltz), INTERVAL '10' SECOND))" +
                "GROUP BY window_start, window_end, sku_id, sku_name");// 按商品分组
tableEnv.toAppendStream(t0, T1.class).addSink(MyClickHouseUtil.getSinkFunction("insert into gmall.Dws_Number_of_additional_purchases values(?,?,?,?,?)"));






        env.execute();
    }
}